package com.nhncorp.pleiades.client;

import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
import org.jgroups.ChannelException;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;

import com.nhncorp.pleiades.common.Constants;
import com.nhncorp.pleiades.conf.PleiadesConfiguration;
import com.nhncorp.pleiades.ens.EventMessage;
import com.nhncorp.pleiades.ens.SpaceMessage;
import com.nhncorp.pleiades.fs.PleiadesFileSystem;
import com.nhncorp.pleiades.hadoop.New_RPC;
import com.nhncorp.pleiades.hadoop.WritableUtils;
import com.nhncorp.pleiades.persistence.LockStatus;
import com.nhncorp.pleiades.protocol.ClientLockInfo;
import com.nhncorp.pleiades.protocol.ClientSessionProtocol;
import com.nhncorp.pleiades.protocol.EventMetaData;
import com.nhncorp.pleiades.protocol.EventServiceProtocol;
import com.nhncorp.pleiades.protocol.LockMetaData;
import com.nhncorp.pleiades.protocol.LockServiceProtocol;
import com.nhncorp.pleiades.protocol.PleiadesInfoData;
import com.nhncorp.pleiades.util.FileUtil;
import com.nhncorp.pleiades.util.NetworkUtil;
import com.nhncorp.pleiades.util.Sleeper;
import com.nhncorp.pleiades.util.StopWatch;

public class LockServiceClient implements LockService, Runnable {
  
  public static final Log LOG = LogFactory.getLog(LockServiceClient.class
      .getName());  
  private LockServiceProtocol lockService = null;
  private EventServiceProtocol eventService = null;
  private ClientSessionProtocol sessionService = null;
  protected PleiadesConfiguration conf;
  private PleiadesUser userInfo = null;
  private PleiadesFileSystem fs;
  private String masterPleiades = "";
  private String[] allPleiades = null;
  private String masterEventManager = "";
  private String[] slaveEventManagers = null;
  private ClientLockInfo lastLockInfo;
  private Thread sessionThread;
  private Thread notificationChannelThread;
  private JChannel notificationChannel = null;
  protected final AtomicBoolean splitBrainMode = new AtomicBoolean(false);
  protected final AtomicBoolean stopSession = new AtomicBoolean(false);
  protected final AtomicBoolean stopNotification = new AtomicBoolean(false);
  private Sleeper sleeper;
  LockEventHandler eventHandler = null;
  EnumSet<Events> eventMask = null;
  private String notificationChannelName = "";
  private String dfsRoot = "";
  protected String serviceId = "";
  private boolean isReadyForDfsAccess = false;
  private Random slaveSelector;
  private final LinkedList<EventMessage> eventQueue = new LinkedList<EventMessage>();
  private long sessionExpiredTimeout;
  private int heartbeatPort = 0;
  private Thread heartbeatResponseServer;
  private final Object lockServiceMonitor = new Object();
  private boolean eventDeliveryMulticast = true;
  
  AtomicBoolean notificationChannelStarted = new AtomicBoolean(false);
  
  public LockServiceClient(PleiadesConfiguration conf) throws IOException {
    this(new PleiadesUser("default"), conf);
  }
  
  public LockServiceClient(PleiadesUser userInfo, PleiadesConfiguration conf) throws IOException {
    this(userInfo, conf, false);
  }
  
  public LockServiceClient(PleiadesUser userInfo, PleiadesConfiguration conf, 
      boolean heartbeatServer) throws IOException {
    System.setProperty("java.net.preferIPv4Stack", "true");    

    this.conf = conf;    
    this.userInfo = userInfo;

    this.sleeper = new Sleeper(1000, this.stopSession);
    this.fs = PleiadesFileSystem.get(conf);
    allPleiades = conf.getStrings("pleiades.cell.members");
    if(allPleiades == null) {
      LOG.error("Can't get pleiades cell info check pleiades.cell.members option in pleiades-site.xml");
      throw new IOException("Can't get pleiades cell info check pleiades.cell.member option in pleiades-site.xml");
    }

    slaveSelector = new Random(System.nanoTime());
    
    boolean foundReacheablePleiades = false;
    for(String pleiades : allPleiades) {
      try {
        InetAddress pleiadesAddress = InetAddress.getByName(pleiades);
        if(isReachable(pleiadesAddress, 8000, 5000)) {
          LOG.debug("ip address for cluster manager : " + pleiades + " is reachable");
          lookupPleiadesCell(pleiades);
          foundReacheablePleiades = true;
          break;
        } 
      } catch (Exception e) {
      }
    }
    if(!foundReacheablePleiades || this.masterPleiades == null || this.masterPleiades.equals("")) {
      String strPleiades = "";
      if(allPleiades != null) {
        for(String entry : allPleiades) {
          strPleiades += entry + " ";
        }
      } 
      throw new IOException("did not find reacheable master pleiades among these servers " + strPleiades);
    }

    lastLockInfo = new ClientLockInfo(this.userInfo.getHostName(), "");
    sessionExpiredTimeout = conf.getLong("pleiades.sessionExpired.timeout", 10*1000);      
    if(!this.userInfo.isSessionDisable()) {
      startSessionMaintainer(heartbeatServer);  
    }
    sleeper.sleep();
  }

  private boolean isReachable(InetAddress entry, int port, int timeout) {
    Socket s = new Socket();
    try {
      s.setSoTimeout(timeout);
      s.connect(new InetSocketAddress(entry, port));
      if(port == heartbeatPort) {
        DataOutputStream out = new DataOutputStream(s.getOutputStream());
        WritableUtils.writeString(out, "R");
        out.flush();
      }
      s.close();
    } catch (Exception e) {
      return false;
    }
    return true;
  }
  
  private void startSessionMaintainer(boolean heartbeatServer) {
    this.stopSession.set(false);
    sessionThread = new Thread(new SessionMaintainer(heartbeatServer), "session Thread");
    sessionThread.setDaemon(true);
    sessionThread.start();
  }
  
  private void startNotificationChannel() throws IOException {
    if (notificationChannelStarted.getAndSet(true) == true) {
      return;
    }
    try {
      this.stopNotification.set(false);
      notificationChannelThread = new Thread(new NotificationChannel(), "Notification Channel Thread");
      notificationChannelThread.setDaemon(true);
      notificationChannelThread.start();
    } catch (Exception e) {
      IOException err = new IOException(e.getMessage());
      err.initCause(e);
      throw err;
    }
  }
  
  private void lookupPleiadesCell(String address) throws IOException {

    this.masterPleiades = address + ":" + conf.get("clusterManager.port", String
        .valueOf(ClientConstants.DEFAULT_LOCK_PORT));
    
    lockService = getMasterPleiadesProxy();
    if(lockService != null) {
      this.masterPleiades = lockService.getMasterPleiades();
      this.serviceId = lockService.getServiceId(userInfo.getServiceName());
      PleiadesInfoData pleInfo = lockService.getPleiadesInfo();
      if(pleInfo != null) {
        notificationChannelName = pleInfo.getNotificationChannel();
        dfsRoot = pleInfo.getDfsRoot();
        splitBrainMode.set(pleInfo.isSplitBrainMode());
        if(!dfsRoot.equals(""))
          isReadyForDfsAccess = true;
        heartbeatPort = pleInfo.getHeartbeatPort();
        eventDeliveryMulticast = pleInfo.isEventUsingMulticast();
      } else {
        LOG.fatal("failed to get pleiades cell information");
        throw new IOException("failed to get pleiades cell information");
      }
    } else {
      throw new IOException("lock service is not ready " + this.masterPleiades);
    }
  }

  public void setHostName(String name) {
    lastLockInfo.setCreator(name);
  }
  
  public void setPath(String path) {
    lastLockInfo.setLockId(path);
  }
  
  public void setLockMode(LockMode mode) {
    lastLockInfo.setLockMode(mode.ordinal());
  }
  
  public void setEventHandler(LockEventHandler handler) throws IOException {    
    if(handler instanceof NativeLockEventHandler) {
      setEventHandler(handler, 0);
    } else {
      setEventHandler(handler, null);  
    }
  }   
  
  public void setEventHandler(LockEventHandler handler, EnumSet<Events> mask) throws IOException {
    
    startNotificationChannel();
    this.eventHandler = handler;
    if(mask == null) {
      this.eventMask = EnumSet.range(Events.CONTENTS_CHANGED, Events.CLIENT_EXPIRED);
    } else {
      this.eventMask = mask;
    }
  }
          
  public void setEventHandler(LockEventHandler handler, int mask)
      throws IOException {
    startNotificationChannel();
    this.eventHandler = handler;
    if (mask == 0) {
      this.eventMask = EnumSet.range(Events.CONTENTS_CHANGED,
          Events.CLIENT_EXPIRED);
    } else {
      this.eventMask = EnumSet.noneOf(Events.class);
      if ((mask & 1) != 0)
        this.eventMask.add(Events.CONTENTS_CHANGED);
      if ((mask & 2) != 0)
        this.eventMask.add(Events.NODE_ADDED);
      if ((mask & 4) != 0)
        this.eventMask.add(Events.NODE_REMOVED);
      if ((mask & 8) != 0)
        this.eventMask.add(Events.NODE_MODIFIED);
      if ((mask & 16) != 0)
        this.eventMask.add(Events.MASTER_FAILOVER);
      if ((mask & 32) != 0)
        this.eventMask.add(Events.SESSION_WARNING);
      if ((mask & 64) != 0)
        this.eventMask.add(Events.SESSION_SAFE);
      if ((mask & 128) != 0)
        this.eventMask.add(Events.SESSION_EXPIRED);
      if ((mask & 256) != 0)
        this.eventMask.add(Events.LOCK_EXPIRED);
      if ((mask & 512) != 0)
        this.eventMask.add(Events.CLIENT_EXPIRED);
    }
  }
  
  public LockEventHandler getEventHandler() {
    return this.eventHandler;
  }    
  
  public boolean exist(String path) throws IOException{
    FileUtil.validate(path);
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling exist:" + path);
    }
    
    return lockService.nodeExist(addServiceId(this.serviceId, path));
    
  }

  public LockMetaData getMetaData(String path) throws IOException {
    FileUtil.validate(path);
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling getMetaData:" + path);
    }
    LockMetaData metaData = lockService.getMetaData(addServiceId(this.serviceId, path));
    if(metaData != null) {
      String userPath = removeServiceId(this.serviceId, metaData.getLockId());
      metaData.setLockId(userPath);
    }
    return metaData;
  }
  
  public LockMetaData[] getMetaDataList(String path, boolean includeDirectory) throws IOException {
    LockMetaData[] metaDataList = null;
    FileUtil.validate(path);
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling getMetaDataList:" + path);
    }
    metaDataList = lockService.readDir(addServiceId(this.serviceId, path), includeDirectory);
    if(metaDataList != null) {
      String userPath;
      for(LockMetaData metaData : metaDataList) {
        userPath = removeServiceId(this.serviceId, metaData.getLockId());
        metaData.setLockId(userPath);
      }
    }
    return metaDataList;
  }  
  
  public Map<String, LockMetaData> readDir(String path) throws IOException {
    return readDir(path, false);
  }
  
  public Map<String, LockMetaData> readDir(String path, boolean includeDirectory) throws IOException {
    Map<String, LockMetaData> metaDataMap = new HashMap<String, LockMetaData>();
    LockMetaData[] metaDataList = getMetaDataList(path, includeDirectory);
    if(metaDataList == null) {
      return null;
    }
    for(LockMetaData entry : metaDataList) {
      metaDataMap.put(entry.getLockId(), entry);
    }
    return metaDataMap;    
  }
  
  public void poison() {
    poison(false);
  }
  
  private void poison(boolean stopNotification) {
    this.stopSession.set(true);
    sleeper.sleep();
    if(sessionThread != null ) {
      sessionThread.interrupt();
    }
    if(stopNotification) {
      this.stopNotification.set(true);
      if(notificationChannelThread != null) {
        notificationChannelThread.interrupt();
      }
      if(heartbeatResponseServer != null) {
        heartbeatResponseServer.interrupt();
      }
    }
  }
  
  public void close() {
    try {
      releaseAll();
    } catch (IOException e) {
      LOG.error("Erro while close:" + e.getMessage(), e);
    }
    shutdown();
  }
  
  public void restartSession() {
    restartSession(false);
  }
  
  private void restartSession(boolean startNotification) {
    try {
      startSessionMaintainer(true);
      if(startNotification) {
        startNotificationChannel();  
      }
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
  
  public boolean deleteNode(String path) throws IOException {
    return deleteNode(path, false);
  }  

  public boolean deleteNode(String path, boolean recursive) throws IOException {
    FileUtil.validate(path);
    ClientLockInfo lockInfo = new ClientLockInfo(userInfo.getCreatorName(),
        addServiceId(this.serviceId, path));
    lockInfo.setLockMode(LockMode.LOCK_EXCLUSIVE.ordinal());
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling deleteNode:" + path);
    }
    
    return lockService.deleteNode(lockInfo, recursive);
  }
  
  public boolean createNode(LockMetaData metaData) throws IOException {
    return createNode(metaData, false, false);
  }
  
  public boolean createNode(LockMetaData metaData, boolean overWrite) throws IOException {
    return createNode(metaData, overWrite, false);
  }
  
  public boolean createNode(LockMetaData metaData, boolean overWrite, boolean isDirectory) throws IOException {
    if(metaData == null) {
      throw new IOException("while calling createNode: metaData is null");
    }
    FileUtil.validate(metaData.getLockId());
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling createNode:" + metaData.getLockId());  
    } 
    
    String path = metaData.getLockId();
    metaData.setLockId(addServiceId(this.serviceId, path));
    metaData.setCreator(userInfo.getCreatorName());
    metaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());
    
    return lockService.createNode(metaData, overWrite, isDirectory);
  }

  public String getContents(String path) throws IOException {
    FileUtil.validate(path);
    ClientLockInfo lockInfo = new ClientLockInfo(userInfo.getCreatorName(),
        addServiceId(this.serviceId, path));
    lockInfo.setLockMode(LockMode.LOCK_SHARED.ordinal());
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling getContents:" + path);
    }
    return lockService.getContents(lockInfo);
  }
  
  public boolean setContents(String path, String contents) throws IOException {
    FileUtil.validate(path);
    ClientLockInfo lockInfo = new ClientLockInfo(userInfo.getCreatorName(), 
        addServiceId(this.serviceId, path));
    lockInfo.setLockMode(LockMode.LOCK_EXCLUSIVE.ordinal());
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling setContents:" + path);
    }
    return lockService.setContents(lockInfo, contents);
  }
  
  public InputStream dfs_open(String path) {
    try {
      if(!isReadyForDfsAccess) {
        LOG.fatal("DFS is not ready");
        return null;
      }
      if(getLock(path, LockMode.LOCK_SHARED)) {
        return fs.open(new Path(dfsRoot + path));  
      }
    } catch (Exception e) {
        e.printStackTrace();
    } 
    return null;
  }
  
  public OutputStream dfs_create(String path) {
    try {
      if(!isReadyForDfsAccess) {
        LOG.fatal("DFS is not ready");
        return null;
      }
      if(getLock(path, LockMode.LOCK_EXCLUSIVE)) {
        return fs.create(new Path(dfsRoot + path));  
      }
    } catch (Exception e) {
        e.printStackTrace();
    } 
    return null;
  }
  
  public boolean dfs_delete(String path) {
    try {
      if(!isReadyForDfsAccess) {
        LOG.fatal("DFS is not ready");
        return false;
      }
      if(getLock(path, LockMode.LOCK_EXCLUSIVE)) {
        return fs.delete(new Path(dfsRoot + path));  
      }
    } catch (Exception e) {
        e.printStackTrace();
    }    
    return false;
  }
  
  public boolean tryLock(String path, LockMode mode) throws IOException {
    ClientLockInfo lockInfo = new ClientLockInfo(userInfo.getCreatorName(), 
        addServiceId(this.serviceId, path));
    lockInfo.setLockMode(mode.ordinal());
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling tryLock:" + path);
    }
    
    return lockService.tryLock(lockInfo);
  }
  
  public boolean tryLock(String path, short mode) throws IOException {
    if(mode == 1) {
      return tryLock(path, LockMode.LOCK_EXCLUSIVE);
    } else {
      return tryLock(path, LockMode.LOCK_SHARED);
    }
  }
  
  public boolean getLock(String path, LockMode mode) throws IOException {
    ClientLockInfo lockInfo = new ClientLockInfo(userInfo.getCreatorName(),
        addServiceId(this.serviceId, path));
    lockInfo.setLockMode(mode.ordinal());
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling getLock:" + path);
    }
    
    String lockResult = lockService.lock(lockInfo);
    if( lockResult.equals("true") ) {
      return true;
    } else if(lockResult.equals("false")){
      return false;
    } else {
      masterPleiades = lockResult;
      if(connectPleiadesCell().lock(lockInfo).equals("true")) {
        return true;
      }
    }
    return false;
  }
  
  public boolean getLock(String path, byte[] metaData, LockMode mode) throws IOException {
    if(metaData != null && metaData.length > ClientLockInfo.MAX_META) {
      throw new IOException("Too big meta value:" + metaData.length + " > " + ClientLockInfo.MAX_META);
    }    
    ClientLockInfo lockInfo = new ClientLockInfo(userInfo.getCreatorName(),
        addServiceId(this.serviceId, path));
    lockInfo.setLockMode(mode.ordinal());
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling getLock(MetaData):" + path);
    }
    String lockResult = lockService.lock(lockInfo);
    if( lockResult.equals("true") ) {
      return true;
    } else if(lockResult.equals("false")){
      return false;
    } else {
      masterPleiades = lockResult;
      if(connectPleiadesCell().lock(lockInfo).equals("true")) {
        return true;
      }
    }
    return false;
  }
  
  public boolean getLock(String path, short mode) throws IOException {
    if(mode == 1) {
      return getLock(path, LockMode.LOCK_EXCLUSIVE);
    } else {
      return getLock(path, LockMode.LOCK_SHARED);
    }
  }
  
  public LockStatus getLockStatus(String path) throws IOException {
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling getLockStatus:" + path);
    }
    return lockService.getLockStatus(addServiceId(this.serviceId, path));
  }
  
  public boolean releaseLock(String path, LockMode mode) throws IOException {
    ClientLockInfo lockInfo = new ClientLockInfo(userInfo.getCreatorName(),
        addServiceId(this.serviceId, path));
    lockInfo.setLockMode(mode.ordinal());
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is not ready while calling releaseLock:" + path);
    }
    String lockResult = lockService.unlock(lockInfo);
    if( lockResult.equals("true") ) {
      return true;
    } else if(lockResult.equals("false")){
      return false;
    } else {
      masterPleiades = lockResult;
      if(connectPleiadesCell().unlock(lockInfo).equals("true")) {
        return true;
      }
    }
    
    return false;
  }
  
  public boolean releaseLock(String path, short mode) throws IOException {
    if(mode == 1) {
      return releaseLock(path, LockMode.LOCK_EXCLUSIVE);
    } else {
      return releaseLock(path, LockMode.LOCK_SHARED);
    }
  }
  
  public boolean releaseAll() throws IOException {
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling releaseAll:");
    }
    return lockService.releaseAllLocks(userInfo.getCreatorName());      
  }
  
  public boolean lockExist(String path) throws IOException {
    boolean isExist = false;
    try {
      lockService = connectPleiadesCell();
      if(lockService == null) {
        throw new IOException("Lock Service is null while calling lockExist:" + path);
      }
      isExist = lockService.lockExist(addServiceId(this.serviceId, path));
    } catch (Exception e) {
      throw new IOException(e.getMessage() + path);
    }
    return isExist;
  }
  
  public LockStatus[] listLock(String path) throws IOException {
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling listLock:" + path);
    }
    LockStatus[] lockList = lockService.listLock(addServiceId(this.serviceId, path));
    if(lockList != null && lockList.length > 0) {
      for(int index=0; index < lockList.length; ++index) {
        String tempLockId = lockList[index].getLockId();
        lockList[index].setLockId(removeServiceId(this.serviceId, tempLockId));
      }
    }
    return lockList;
  }
  
  public String getAllLockStatus() throws IOException {
    lockService = connectPleiadesCell();
    if(lockService == null) {
      throw new IOException("Lock Service is null while calling getAllLockStatus:");
    } 
    return lockService.getLockDump();
  }
  
  public boolean subscribe(String type, String id) {
    boolean isSuccess = false;
    try {
      EventMetaData metaData = new EventMetaData(type, userInfo.getHostName());
      EventServiceProtocol eventService = connectEventManagerCell();
      metaData.setOwnerId(this.userInfo.getHostName());
      metaData.setEventKey(id);
      eventService.register(metaData, 0);
      isSuccess = true;
      
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      if(!isSuccess)
        return isSuccess;
    }
    return isSuccess;
  }
  
  public boolean writeEntry(SpaceEntry entry) {
    boolean isSuccess = false;
    try {
      EventMetaData metaData = new EventMetaData(entry.getType(), userInfo.getHostName());
      EventServiceProtocol eventService = connectEventManagerCell();
      metaData.setEventKey(entry.getKey());
      eventService.write(metaData);
      isSuccess = true;
      
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      if(!isSuccess)
        return isSuccess;
    }
    return isSuccess;
  }
  
  public boolean writeEntry(List<SpaceEntry> entryList) {
    boolean isSuccess = false;
    try {
      EventMetaData[] metaDatas = new EventMetaData[entryList.size()];
      EventServiceProtocol eventService = connectEventManagerCell();
      int metaIndex = 0;
      for(SpaceEntry entry : entryList) {
        metaDatas[metaIndex] = new EventMetaData(entry.getType(), userInfo.getHostName());
        metaDatas[metaIndex].setEventKey(entry.getKey());
        ++metaIndex;
      }
      eventService.write(metaDatas);
      isSuccess = true;
      
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      if(!isSuccess)
        return isSuccess;
    }
    return isSuccess;
  }
  
  protected static String addServiceId(String serviceId, String path ) {
    if(path.startsWith("/")) {
      return serviceId + path;
    }
    return serviceId + "/" + path;
  }
  
  private static String removeServiceId(String id, String path ) {
    if(id != null && !id.equals("")) {
      if(path.startsWith(id)) {
        path = path.substring(id.length());
      } else {
        LOG.fatal("lock path does not start with service id : " + id + ", " + path);
      }
    } 
    return path; 
  }
  
  private static String removeServiceName(String name, String path ) {
    if(name != null && !name.equals("")) {
      if(path.startsWith(name)) {
        path = path.substring(name.length() + Constants.SERVICE_DELIMETER.length());
      } else {
        LOG.fatal("owner does not start with service name : " + name + ", " + path);
      }
    } 
    return path; 
  }  
  
  private LockServiceProtocol getMasterPleiadesProxy() throws IOException {
    
    int maxRetryWhenException = 5;
    int retry = 0;
    LockServiceProtocol pleiades = null;
    
    if(allPleiades != null && allPleiades.length > 0) {
      
      for(String entry : allPleiades) {
        LOG.debug("try to update master within servers of pleiades cell: " + entry);
        InetAddress pleiadesAddress = InetAddress.getByName(entry);
        if(!isReachable(pleiadesAddress, 8000, 5000)) {
          LOG.info(pleiadesAddress + " is not reachable");
          continue;
        }
        
        InetSocketAddress address = NetworkUtil.getAddress(entry + ":" + conf.get("clusterManager.port", String
            .valueOf(ClientConstants.DEFAULT_LOCK_PORT)));
        retry = 0;
        while (retry < maxRetryWhenException) {
          try {
            pleiades = (LockServiceProtocol) New_RPC.getProxy(
                LockServiceProtocol.class, LockServiceProtocol.versionID, address, conf);
            if(pleiades.isPrimaryMaster()) {
              return pleiades;
            } else {
              LOG.info(entry + " is not master of primary partition");
              pleiades = null;
            }
            break;
          } catch (Exception e) {
            LOG.warn("Error while trying to update master:" + e.getMessage() + ", but retry[" + retry + "]" , e);
            pleiades = null;
            retry++;
            if (retry >= maxRetryWhenException) {
              LOG.error("Error while trying to update master:" + e.getMessage(), e);
              break;
            }
            sleeper.sleepMilliSeconds(100);
          } finally {
          }
        }
      }      
    } 
    
    return null;
  }
  
  private LockServiceProtocol connectPleiadesCell() throws IOException {
    
    long maxRetryWhenException = (sessionExpiredTimeout / 1000);
    int retry = 0;
    while (retry < maxRetryWhenException) {
      try {
        if( lockService == null) {
          LockServiceProtocol pleiadesProxy = getMasterPleiadesProxy();
          if(pleiadesProxy != null) {
            synchronized (lockServiceMonitor) {
              this.lockService = pleiadesProxy;  
            } 
          } else {
            sleeper.sleep();
          }
          ++retry;
          continue;
          
        } else {
          String tempId = lockService.getServiceId(userInfo.getServiceName());
          if(tempId == null || tempId.equals("") || tempId.equals("false")) {
            LOG.fatal("not authorized or not defined service type.  check service_acl.xml file");
            throw new IOException("NOT_AUTHORIZED_USER for service [" + userInfo.getServiceName() + "]");
          } else {
            serviceId = tempId;
            return lockService;
          }
        }
        
      } catch (Exception e) {
        synchronized (lockServiceMonitor) {
          lockService = null;  
        }
        if(e instanceof IOException) {
          if(e.getMessage().indexOf("NOT_AUTHORIZED_USER") >= 0) {
            throw (IOException)e;
          }
        }
        LOG.warn("try " + retry + " got exception while trying to connect pleiades cell " + this.masterPleiades);
        ++retry;
        if (retry >= maxRetryWhenException - 1) {
          LOG.error("Error:" + e.getMessage(), e);
          IOException err = new IOException(e.getMessage());
          err.initCause(e);
          throw err;
        }
        sleeper.sleep();
      }
    }
    
    throw new IOException("can't get lock service");
  }
  
  private ClientSessionProtocol connectSessionCell(String server) throws IOException {
    if(server == null || server.trim().length() == 0) {
      throw new IOException("connectSessionCell: No server info");
    }
    
    int maxRetry = 10;
    int retry = 0;
    while (retry < maxRetry) {
      try {
        if (sessionService == null) {
          sessionService = (ClientSessionProtocol) New_RPC.getProxy(
              ClientSessionProtocol.class, ClientSessionProtocol.versionID,
              NetworkUtil.getAddress(server), conf);
        } else {
          return sessionService;
        }
      } catch (Exception e) {
        LOG.warn("try " + retry + " got exception while trying to connectSessionCell " + server);
        sessionService = null;
        retry++;
        if (retry >= maxRetry) {
          LOG.error("got exception while trying to connectSessionCell " + server, e);
          break;
        }
        sleeper.sleep();
      }
    }
    throw new IOException("can't get session cell");
  }
  
  private EventServiceProtocol connectEventManagerCell() throws IOException {
    return connectEventManagerCell(true);
  }
  
  private EventServiceProtocol connectEventManagerCell(boolean master) throws IOException {
    int maxRetry = 10;
    int retry = 0;
    while (retry < maxRetry) {
      try {
        if( eventService == null) {
          String eventManager = "";
          if(master) {
            eventManager = masterEventManager;
          } else {
            eventManager = slaveEventManagers[Math.abs(slaveSelector
                .nextInt(slaveEventManagers.length))];
          }
          eventService = (EventServiceProtocol) New_RPC.getProxy(
              EventServiceProtocol.class, EventServiceProtocol.versionID,
              NetworkUtil.getAddress(eventManager), conf);
        }
        break;
      } catch (Exception e) {
        LOG.debug("try " + retry + " got exception while trying to connect event manager");
        eventService = null;
        retry++;
        if (retry >= maxRetry) {
          e.printStackTrace(System.out);
          break;
        }
        sleeper.sleep();
      }
    }
    return eventService;
  }
  
  class SessionMaintainer implements Runnable {
    String sessionServer = "";
    List<String> sessionServerList = new ArrayList<String>();
    StopWatch timer;
    Socket socket = null;
    InetSocketAddress targetAddr = null;
    int heartbeatInterval = 5*1000;
    boolean heartbeatServer;
    
    public SessionMaintainer(boolean heartbeatServer) {
      heartbeatInterval = conf.getInt("pleiades.client.heartbeat.interval", 5*1000);
      this.heartbeatServer = heartbeatServer;
      
      for(String slave : allPleiades) {
        int ipLength = slave.indexOf(":");
        if( ipLength > 0) {
          slave = slave.substring(0, ipLength);
        }
        sessionServerList.add(slave);
      }
    }
    
    public void run() {
      LOG.debug("SessionMaintainer:" + heartbeatServer);
      if(heartbeatServer) {
        String hostName = userInfo.getHostName();
        if(hostName == null || userInfo.getHeartbeatPort() == 0) {
          LOG.warn("No Heartbeat receiver info:host=" + hostName + ",port=" + userInfo.getHeartbeatPort());
          return;
        }
        String bindAddress = "0.0.0.0";
        
        try {
          ServerSocketChannel ssChannel = ServerSocketChannel.open();
          ServerSocket ss = ssChannel.socket();
          ss.bind(new InetSocketAddress(bindAddress, userInfo.getHeartbeatPort()));
          ss.setReuseAddress(true);
          heartbeatResponseServer = new Thread(new HeartbeatResponseServer(ssChannel));  
          heartbeatResponseServer.setDaemon(true);
          heartbeatResponseServer.start();
          LOG.debug("create heartbeat socketserver thread:" + userInfo.getHeartbeatPort());
        } catch (IOException ie) {
          LOG.error("Could not open HeartSocketServer server at " + userInfo.getHeartbeatPort(), ie);
        }
      }
      
      LOG.debug("---------- session thread started ----------");
      try {

        sessionServer = getReachableSessionServer(splitBrainMode.get());
        if(sessionServer != null && !sessionServer.equals("")) {
          targetAddr = NetworkUtil.getAddress(sessionServer + ":" + String.valueOf(heartbeatPort));  
        } else {
          throw new IOException("live session server list is empty");
        }
        
        LOG.info("initially connected to session server: " + sessionServer);
     
      } catch (Exception e) {
        LOG.warn("Error SessionMaintainer.run():" + e.getMessage(), e);
        // maybe insert code return;
      }
      loopForSession();
    }
    
    private void loopForSession() {
      AtomicInteger intervalChecker = new AtomicInteger(1);
      timer = new StopWatch();
      while(!stopSession.get()) {
        try {
          if(timer.elapsedTime() > (sessionExpiredTimeout + 1000)) {
            LOG.fatal(userInfo.getHostName() + " elapsed time is "
                + timer.elapsedTime() + " since the last heartbeat");
            if (eventHandler != null) {
              synchronized (eventQueue) {
                eventQueue.add(new EventMessage(userInfo.getCreatorName()
                    , LockService.Events.SESSION_EXPIRED.ordinal(), 
                    "", false, System.currentTimeMillis()));
                eventQueue.notifyAll();
              }          
            }
          }
          socket = new Socket();
          socket.setSoLinger(true, 1);
          socket.setSoTimeout(5*1000);
          socket.connect(targetAddr, 5*1000);
          DataOutputStream out = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream()));
          WritableUtils.writeString(out, userInfo.getCreatorName());
          out.flush();
          socket.close();
          
          if (splitBrainMode.get()) {
            if (intervalChecker.getAndIncrement() >= conf.getInt("pleiades.cell.partitionCheckInterval", 3)) {      
              intervalChecker.set(1);
              String primaryPartitionServer = "";
              primaryPartitionServer = getReachableSessionServer(true);
              if(primaryPartitionServer != null && !primaryPartitionServer.equals("")) {
                if(!sessionServer.equals(primaryPartitionServer)) {
                  targetAddr = NetworkUtil.getAddress(primaryPartitionServer + ":" + String.valueOf(heartbeatPort));  
                } 
              } else {
                throw new IOException("live primary-partitioned session server list is empty");
              }
            }
          }
          timer.reset();
        } catch (Exception e) {
          
          LOG.warn("Exception!!!!", e);
          if(!sessionServerList.isEmpty()) {
            
            sessionServer = getReachableSessionServer(splitBrainMode.get());
            if(sessionServer != null && !sessionServer.equals("")) {
              targetAddr = NetworkUtil.getAddress(sessionServer + ":" + String.valueOf(heartbeatPort));
              LOG.info("connected to session server: " + sessionServer);           
            } else {
              LOG.fatal("currently no live session server and should wait until session servers are recovered ...");
              sleeper.sleep();
            }
          } else {
            LOG.fatal("pleiades.cell.master and members were misconfigured and added master plieades to session server list");
            sessionServerList.add(masterPleiades);
          }
        } finally {
          try {
            if(socket != null && !socket.isClosed()) {
              socket.close();  
            }
            sleeper.sleepMilliSeconds(heartbeatInterval);
          } catch (IOException e) {
          }
        }
      }
    }
    
    private String getReachableSessionServer(boolean primaryPartition) {
       
      String masterSessionServer = "";
      String reacheableServer = "";
      List<String> liveSlaveSessionServers = new ArrayList<String>();
      for(String entry : sessionServerList) {
        try {
          if(isReachable(InetAddress.getByName(entry), heartbeatPort, 3000)) {
            if(!masterPleiades.equals("") && masterPleiades.startsWith(entry)) {
              masterSessionServer = entry;
              continue;
            } 
            liveSlaveSessionServers.add(entry);
          }
        } catch (UnknownHostException ue) {}
      }
      
      if(primaryPartition) {
        List<String> primaryPartitionedServers = new ArrayList<String>();
        if(!liveSlaveSessionServers.isEmpty()) {
          for(String server : liveSlaveSessionServers ) {
            /** add the code right here to check whether connected server is primary partition 
             * **/
            primaryPartitionedServers.add(server);
          }
          if(!primaryPartitionedServers.isEmpty()) {
            reacheableServer = primaryPartitionedServers.get(Math.abs(slaveSelector.nextInt(primaryPartitionedServers.size())));  
          }
        } else if (!masterSessionServer.equals("") ) {
          /** add the code right here to check whether connected server is primary partition 
           * **/
          reacheableServer = masterSessionServer;
        } else {
          LOG.fatal("serious problem : live session server list is empty");
        }
        
      } else {
        if(!liveSlaveSessionServers.isEmpty()) {
          reacheableServer = liveSlaveSessionServers.get(Math.abs(slaveSelector.nextInt(liveSlaveSessionServers.size())));
        } else if (!masterSessionServer.equals("") ) {
          reacheableServer = masterSessionServer;
        } else {
          LOG.fatal("serious problem : live session server list is empty");
        }
      }
      
      return reacheableServer;
    }

    class HeartbeatResponseServer implements Runnable {
      ServerSocketChannel ssChannel;
      public HeartbeatResponseServer(ServerSocketChannel ssChannel) {
          this.ssChannel = ssChannel;
      }

      public void run() {
        try {
          while(!stopSession.get()) {
            //LOG.debug("Start heartbeat socketserver thread");
            if(ssChannel == null) {
              break;
            }
            try {
              SocketChannel socketChannel = ssChannel.accept();
              Socket socket = socketChannel.socket();
              OutputStream heartbeatOut = null;
              try {
                heartbeatOut = socket.getOutputStream();
                heartbeatOut.write("OK".getBytes());
                heartbeatOut.flush();
              } finally {
                if(heartbeatOut != null) {
                  heartbeatOut.close();
                }
              }
              if(timer != null) {
                timer.reset();
              }
            } catch (ClosedByInterruptException closeE) {
              if(ssChannel != null) {
                if(ssChannel.isOpen()) {
                  LOG.warn("SocketServerChannel not closed");
                } else {
                  ssChannel.close();
                  ssChannel = null;
                }
              }
              LOG.warn("HeartbeatResponseServer successfuly closed by Interrupt");
              return;
            } catch (IOException ie) {
              LOG.warn("Error while receive from cluster manager: " + ie.getMessage());
            }
          }
          ssChannel.close();
          ssChannel = null;
          LOG.warn("HeartbeatResponseServer successfuly closed");
        } catch (IOException ie) {
            LOG.info("Exiting HeartbeatResponseServer due to " + ie.toString());
        }
      }
    }
  }
  
  class NotificationChannel extends ReceiverAdapter implements Runnable {

    public List<EventFetcher> eventFetcherList;

    public NotificationChannel() throws ChannelException {
      if(eventDeliveryMulticast) {
        notificationChannel = new JChannel(Constants.NOTIFICATION_ADDRESS + Constants.CLIENT_PROPERTY);
        
      } else {
        String localIpAddress = "";
        String property = "";
        try {
          String localHostName = InetAddress.getLocalHost().getCanonicalHostName();
          localIpAddress = InetAddress.getByName(localHostName).getHostAddress();
        } catch (IOException e) {
          e.printStackTrace();
        }
        String pleiadesHostNames = "";
        for (String entry : allPleiades) {
          pleiadesHostNames += (entry + "[7800],");
        }
        if (pleiadesHostNames.endsWith(",")) {
          pleiadesHostNames = pleiadesHostNames.substring(0, pleiadesHostNames.length() - 1);
          property = Constants.TCP_NIO_PROPERTY.replaceFirst(
              "hosts=localhost\\[7800\\]", "hosts=" + pleiadesHostNames);
        } else {
          property = Constants.TCP_NIO_PROPERTY;
        }
        if (localIpAddress != null && !localIpAddress.equals("")
            && !localIpAddress.equals("127.0.0.1")) {
          property = property.replaceAll("localhost", localIpAddress);
        } else {
          LOG.info("failed to get real ip address and could cause join operation to fail");
        }
        property = property.replaceFirst("port_range=1", "port_range=100");
        property = property.replaceFirst("disable_initial_coord=false",
            "disable_initial_coord=true");
        notificationChannel = new JChannel(property);
      }
      
      notificationChannel.setReceiver(this);
      notificationChannel.connect(notificationChannelName);
      int handlerCount = conf.getInt("pleiades.client.eventhandler.count", 3);
      eventFetcherList = new ArrayList<EventFetcher>();
      for(int count=0; count < handlerCount; ++count) {
        eventFetcherList.add( new EventFetcher("EventFetcher Thread " + String.valueOf(count)) );        
      }
    }
    
    public void run() {
      LOG.debug("--------------- notification channel thread started from LockServiceClient---------------");
      for(EventFetcher fetcher : eventFetcherList) {
        fetcher.start();
      }
      notificationChannelLoop();
    }

    private void notificationChannelLoop() {
      while (!stopNotification.get()) {
        try {
          Thread.sleep(100);
        } catch (InterruptedException e) {
          notificationChannel.close();
          notificationChannel.shutdown();
          for(EventFetcher fetcher : eventFetcherList) {
            fetcher.interrupt();
          }
          break;
        }
      }
      notificationChannel.close();
      notificationChannel.shutdown();
    }

    public void receive(Message msg) {
      
      if (msg.getObject() instanceof EventMessage) {
        EventMessage eventMessage = (EventMessage) msg.getObject();
        if (eventHandler != null && eventMessage != null) {
          synchronized (eventQueue) {
            eventQueue.add(eventMessage);
            eventQueue.notifyAll();
          }          
        }
        LOG.debug("received event message " + eventMessage
            + " from lock service client");
      } else if (msg.getObject() instanceof SpaceMessage) {
        SpaceMessage eventMessage = (SpaceMessage) msg.getObject();
        LOG.debug("received space message " + eventMessage
            + " from lock service client");
        
      } else {
        LOG.debug("received unknown type of message and dropped");
      } 
    }
    
    class EventFetcher extends Thread {
      public EventFetcher(String name) {
        super(name);
      }
      public void run() {
        while(!stopNotification.get()) {
          try {                 
            EventMessage[] events = null;
            synchronized(eventQueue) {
              while (eventQueue.isEmpty()) {
                eventQueue.wait();
              }                         
              events = eventQueue.toArray(new EventMessage[]{});
              eventQueue.clear();
            }
            for(EventMessage eventEntry : events) {
              if (!eventEntry.getLockId().equals("")) {
                if (!eventEntry.getLockId().startsWith(serviceId)) {
                  LOG.debug("lock's service id is different and drop event message "
                          + eventEntry);
                  continue;
                }
              }
              if (!eventEntry.getHostName().equals("")) {
                if (!eventEntry.getHostName().startsWith(userInfo.getServiceName())) {
                  if(!(eventEntry.getEventType() == LockService.Events.MASTER_FAILOVER.ordinal())) {
                    LOG.debug("owner's service name is different and drop event message "
                        + eventEntry);
                    continue;                    
                  }
                }
              }
              fireEvent(eventEntry);  
            }
          } catch (InterruptedException e) {
            return;
          }
        }
      }
    };
    
    public void viewAccepted(View view) {
    }
    
    private void fireEvent(EventMessage eventMessage) {
      
      // does not need to wrap EventMessage with LockEventData. remove this in the near future...
      LockEventData eventData = new LockEventData();
      eventData.setEventType(eventMessage.getEventType());
      if(eventMessage.getHostName().equals("")) {
        eventData.setHostname("");  
      } else {
        if(eventData.getEventType() == LockService.Events.MASTER_FAILOVER.ordinal()) {
         eventData.setHostname(eventMessage.getHostName()); 
        } else {
          eventData.setHostname(removeServiceName(userInfo.getServiceName(),
              eventMessage.getHostName()));          
        }
      }
      if(eventMessage.getLockId().equals("")) {
        eventData.setLockId("");
      } else {
        eventData.setLockId(removeServiceId(serviceId, eventMessage.getLockId()));  
      }
      eventData.setDirectory(eventMessage.getIsDirectory());
      eventData.setGenerationNumber(eventMessage.getGenerationNumber());
      
      if(eventMessage.getEventType() == LockService.Events.CONTENTS_CHANGED.ordinal()) {
        if( eventMask.contains(Events.CONTENTS_CHANGED) ) {
          eventHandler.contentsModified(eventData);
        }
      } else if(eventMessage.getEventType() == LockService.Events.NODE_ADDED.ordinal()) {
        if( eventMask.contains(Events.NODE_ADDED) ) {
          eventHandler.nodeAdded(eventData);
        }
      } else if(eventMessage.getEventType() == LockService.Events.NODE_REMOVED.ordinal()) {
        if( eventMask.contains(Events.NODE_REMOVED) ) {
          eventHandler.nodeRemoved(eventData);
        }
      } else if(eventMessage.getEventType() == LockService.Events.NODE_MODIFIED.ordinal()) {
        if( eventMask.contains(Events.NODE_MODIFIED) ) {
          eventHandler.nodeModified(eventData);
        }
      } else if(eventMessage.getEventType() == LockService.Events.MASTER_FAILOVER.ordinal()) {
        if( eventMask.contains(Events.MASTER_FAILOVER) ) {
          LOG.fatal("master server failed over : " + eventMessage.getHostName());
          eventHandler.masterFailover();
        }
      } else if(eventMessage.getEventType() == LockService.Events.SESSION_WARNING.ordinal()) {
        if( eventMask.contains(Events.SESSION_WARNING) ) {
          eventHandler.sessionWarning();
        }
      } else if(eventMessage.getEventType() == LockService.Events.SESSION_SAFE.ordinal()) {
        if( eventMask.contains(Events.SESSION_SAFE) ) {
          eventHandler.sessionSafe();
        }
      } else if(eventMessage.getEventType( )== LockService.Events.SESSION_EXPIRED.ordinal()) {
        if( eventMask.contains(Events.SESSION_EXPIRED) ) {
          eventHandler.sessionExpired();
        }
      } else if(eventMessage.getEventType( )== LockService.Events.LOCK_EXPIRED.ordinal()) {
        if( eventMask.contains(Events.LOCK_EXPIRED) ) {
          eventHandler.lockExpired(eventData);
        }
      } else if(eventMessage.getEventType( )== LockService.Events.CLIENT_EXPIRED.ordinal()) {
        if( eventMask.contains(Events.CLIENT_EXPIRED) ) {
          eventHandler.clientExpired(eventData);
        }
      } else {
        LOG.debug("not pre-defined event message type and dropped");
        return; 
      }
    }
  }
  
  public void shutdown() {
    poison(true);
  }
  
  public void run() {
    
  }
  
  public String toString() {
    return this.userInfo.getHostName();
  }
  
  public static void main(String args[]) {
    try {
      LockServiceClient client = new LockServiceClient(new PleiadesUser(""), new PleiadesConfiguration());
      Thread thread = new Thread(client);
      thread.start();
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
}